package com.hadoop.mr.pagerank;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * <p>pageRank计算的Map</p>
 *
 * @author Andy
 * @date 2017/10/26
 */
public class PageRankMap extends Mapper<Text, Text, Text, Text> {

    private Text outKey = new Text();
    private Text outVal = new Text();

    @Override
    protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
        context.write(key, value);

        Node node = Node.fromMr(value.toString());
        if (node.getAdjacentNodeNames() != null && node.getAdjacentNodeNames().length > 0) {
            double outPageRank = node.getPageRank() / node.getAdjacentNodeNames().length;
            for (String neighbor : node.getAdjacentNodeNames()) {
                outKey.set(neighbor);
                Node adjacentNode = new Node().setPageRank(outPageRank);
                outVal.set(adjacentNode.toString());
                System.out.println("  output -> K[" + outKey + "],V[" + outVal + "]");
                context.write(outKey, outVal);
            }
        }
    }
}
